package com.foton.bpc.all.register.impl;

import com.foton.bpc.all.register.NotifyListener;
import com.foton.bpc.all.register.Register;
import com.foton.bpc.all.util.URIUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.net.URI;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author RenPu
 * @Date 2023/6/10 15:29
 * @Version 1.0
 * @Description: Redis作为注册中心的实现
 **/

@Slf4j
public class RedisRegister implements Register {

    //服务实例续约得时间周期
    private static final int TIME_OUT=15;

    //定时线程池
    ScheduledThreadPoolExecutor scheduledThreadPool=new ScheduledThreadPoolExecutor(10);

    private URI address;

    /**    服务提供者   **/
    //心跳维护得集合
    List<URI> serviceHeartBeat =new ArrayList<>();


     /**    服务消费者   **/
    //本地缓存得集合
    private JedisPubSub jedisPubSub;
    Map<String, Set<URI>> localMap=new HashMap<>();
    //订阅监听得集合
    Map<String,NotifyListener> listenerMap=new ConcurrentHashMap<>();



    /**
     * //注册方法
     * @param uri
     */
    @Override
    public void register(URI uri) {
        String key="bpc-"+uri.toString();
        Jedis jedis = new Jedis(address.getHost(),address.getPort());
        jedis.setex(key,TIME_OUT,String.valueOf(TimeUnit.MILLISECONDS));
        jedis.close();
        //心跳检测开始
        serviceHeartBeat.add(uri);
    }


    /**
     * 获取redis得配置参数。开始进行对serviceHeartBeat，定时续约
     * //init方法 配置链接方法
     * @param address
     */
    @Override
    public void init(URI address) {
        this.address=address;
        scheduledThreadPool.scheduleWithFixedDelay(()->{
            Jedis jedis = new Jedis(address.getHost(),address.getPort());

         try{
             if(!CollectionUtils.isEmpty(serviceHeartBeat)){
                 for (URI uri : serviceHeartBeat) {
                     String key="bpc-"+uri.toString();
                     jedis.expire(key,TIME_OUT);

                 }
             }else {
                 log.info("当前没有需要进行，续约得服务实例！！！！！！");
             }
             jedis.close();
         }catch (Exception e){
             e.printStackTrace();
         }

        },3000,5000, TimeUnit.MILLISECONDS);  //周期5秒，进行续约服务


        //监听服务变动
        scheduledThreadPool.execute(()->{

            jedisPubSub=new JedisPubSub() {
                @Override
                public void onPSubscribe(String pattern, int subscribedChannels) {
                    log.info("注册中心开始监听："+pattern);
                }

                /**
                 *
                 * @param pattern
                 * @param channel
                 * @param message  事件，例如：get set keys等命令事件
                 */
                @Override
                public void onPMessage(String pattern, String channel, String message) {
                    try{
                        URI serviceUri = new URI(channel.replace("_keyspace@0_:bpc-",""));
                        Set<URI> uriSet = localMap.get(URIUtils.getService(serviceUri));
                        if(!CollectionUtils.isEmpty(uriSet)){
                        //新增，监听set指令 维护本地缓存
                        if("set".equals(message)){
                            uriSet.add(serviceUri);
                        }

                        //过期 移除本地缓存
                        if("expired".equals(message)){
                           uriSet.remove(serviceUri);

                        }
                        if("set".equals(message)||"expired".equals(message)){
                            log.info("服务实例存在变化，，开始刷新！！！！");

                            NotifyListener notifyListener = listenerMap.get(URIUtils.getService(serviceUri));
                            if(notifyListener!=null){
                                notifyListener.notify(localMap.get(URIUtils.getService(serviceUri)));
                            }
                        }

                    }

                    }catch (Exception e){
                        e.printStackTrace();
                    }

                    Jedis jedis=new Jedis(address.getHost(),address.getPort());

                    //设置发布订阅得监听命令参数，以及观测key值
                    jedis.subscribe(jedisPubSub,"_keyspace@0_:bpc-*");
                }

            };

        });

    }



    /**
     * //监听方法 监测实例变化
     * @param service  结合redis得发布订阅机制，需要通过配置进行开启  redis.conf notify-keyspace-events KE$xg
     * @param notifyListener
     */
    @Override
    public void subscribe(String service, NotifyListener notifyListener) {

        try {
            //如果本地缓存localMap没有根据，service获取到值
            if (null == localMap.get(service)) {
                localMap.putIfAbsent(service,new HashSet<>());
                listenerMap.putIfAbsent(service,notifyListener);

                //构建redis客户端，根据特殊字符进行模糊匹配，注册中心上村存储得数据是否包含
                // bpc-* +service+ ?*  进行模糊搜索
                Jedis jedis = new Jedis(address.getHost(), address.getPort());
                Set<String> serviceInstances = jedis.keys("bpc-*" + service + "?*");
                for (String instance : serviceInstances) {
                    URI uri = new URI(instance.replace("bpc-",""));
                    localMap.get(service).add(uri);
                }

            }
            //将从注册中心获取得数据。替换掉特殊前缀，存储到localMap本地缓存中
            notifyListener.notify(localMap.get(service));

        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

